#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2023/10/9 21:43
import asyncio
import time
import allure
from google.protobuf.json_format import MessageToJson
from apis.cgsession.access import post_union_apply
from apis.cgsession.servo import post_servo_usertoken
from init_models import EnvInfo
from public_tools.common_vars import log
from public_tools.pb2 import cloudgame_pb2
import json
from public_tools.api_utils.http_client import HttpClient
from public_tools.files_utils.data_util import DataUtil
from public_tools.time_utils.time_util import TimeUtil
from public_tools.assert_utils.assert_util import AssertUtils
import websockets
import ssl
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

class FullLinkRunner:
    def __init__(self, **options):
        self.heartbeat_limit = options.get("heartbeat_limit", 300)
        self.heartbeat_interval = options.get("heartbeat_interval", 5)
        self.gid = options.get("gid", 33887)
        self.biz = options.get("biz", 4)
        self.vm_id = options.get("vmid", 0)
        self.app_user_id = options.get("app_user_id", "13711153040")
        self.token, self.uuid, self.flow_id, self.ws_path = None, None, None, None
        self.servo_host = "https://beta-08-api-bnd.yuntiancloud.com:18113"
        self.access_host = "https://beta-08-api-bnd.yuntiancloud.com:18112"
        self.mode = 0  # 启动模式  1,拨测模式
        self.launch_success = False  # 标记字段    是否启动完成
        self.keep_alive = True  # 标记字段 控制流程是否断开
        self.flag = 0  # 标记字段  2.监控各类事件 pb发送exitgame
        self.finish = 0  # 1.全链路过程完成标志
        self.game_status_ready_render = True

        self.launch_info = {}  # 启动记录，保存各种返回参数
        self.device_info = {
            "board": "cc2",
            "brand": "vivo",
            "cpu": "armeabi-v7a",
            "hardware": "dd2",
            "manufacturer": "CMDC",
            "model": "bb2",
            "network": "Wifi",
            "os_version": "CM201-2",
            "produt": "Hi3798MV300",
            "sdk": "19",
            "dev_id": "abc"
        }  # 可配置项

    def reset(self):
        """
        @return:
        """
        self.launch_success = False  # 标记字段    是否启动完成
        self.keep_alive = True  # 标记字段 控制流程是否断开
        self.flag = 0  # 标记字段  2.监控各类事件 pb发送exitgame
        self.finish = 0  # 1.全链路过程完成标志

        self.launch_info = {}  # 启动记录，保存各种返回参数

        self.device_info = {
            "board": "cc2",
            "brand": "vivo",
            "cpu": "armeabi-v7a",
            "hardware": "dd2",
            "manufacturer": "CMDC",
            "model": "bb2",
            "network": "Wifi",
            "os_version": "CM201-2",
            "produt": "Hi3798MV300",
            "sdk": "19",
            "dev_id": "abc"
        }  # 可配置项

    def get_heart_beat_msg(self):
        """
        心跳消息构造
        :return:
        """
        # 心跳消息
        heartbeat = cloudgame_pb2.CGReqHeartbeat()
        hb_header = heartbeat.header
        hb_header.timestamp = int(TimeUtil.get_current_millisecond())
        hb_header.token = self.token
        hb_header.request_id = DataUtil.random_uuid4()

        cg_message = cloudgame_pb2.CGMessage()
        cg_message.cmdtype = cloudgame_pb2.ReqHeartbeat
        cg_message.version = 6
        cg_message.payload = heartbeat.SerializeToString()
        return cg_message.SerializeToString()

    def get_launch_msg(self):
        """ 消息构造
        @return:
        """
        msg = cloudgame_pb2.CGMessage()
        msg.cmdtype = cloudgame_pb2.ReqLaunchGame
        msg.version = 6

        launch_msg = cloudgame_pb2.CGReqLaunchGame()
        header = launch_msg.header
        header.timestamp = int(TimeUtil.get_current_millisecond())
        header.token = self.token
        header.request_id = DataUtil.random_uuid4()

        stream_info = {"clientSupportHevc": 1, "connectionType": "udpTcp", "maxFps": 60,
                       "policy": "{\"network_info\":{\"bwe_unzero_middle90\":0,\"bwe_unzero_num\":0,\"ctldelay_middle90\":0,\"ctldelay_total_num\":0,\"delay_middle90\":0,\"delay_total_num\":0,\"effective\":0,\"loss_total_num\":0,\"loss_unzero_middle90\":0,\"loss_unzero_num\":0,\"losstype_counts\":null,\"losstype_total_num\":0},\"policy_info\":{\"effective\":0,\"enable_disable_fec\":1,\"fec_default_rate\":20,\"fec_low_rtt_nack_ms\":10,\"max_bandwidth\":3000,\"min_bandwidth\":1000,\"start_bandwidth\":2000}}",
                       "requestUseServerCursorData": True}
        launch_msg.game_id = self.gid
        launch_msg.resource_type = cloudgame_pb2.ANY
        launch_msg.opaque = 'just test'
        launch_msg.vmid = self.vm_id
        launch_msg.device_info = json.dumps(self.device_info)
        launch_msg.stream_info = json.dumps(stream_info)

        msg.payload = launch_msg.SerializeToString()

        msg_text = msg.SerializeToString()
        allure.attach(f"flowId({self.flow_id}), uuid({self.uuid}) launch msg({msg_text})", name = "stream logs", attachment_type = allure.attachment_type.TEXT)
        log.info(f'flowId({self.flow_id}), uuid({self.uuid})  launch msg({msg_text})')
        return msg_text

    def get_alloc_vpad_msg(self):
        """
        @return:
        """
        msg = cloudgame_pb2.CGMessage()
        msg.cmdtype = cloudgame_pb2.ReqAllocVPad
        msg.version = 6

        alloc_vpad = cloudgame_pb2.CGReqAllocVPad()
        header_pad = alloc_vpad.header
        header_pad.timestamp = int(TimeUtil.get_current_millisecond())
        header_pad.token = self.token
        header_pad.request_id = DataUtil.random_uuid4()
        alloc_vpad.vpad_id = 2
        alloc_vpad.dpad_id = 0
        # AllocVPad.dpad_id = int('0')
        msg.payload = alloc_vpad.SerializeToString()
        msg_text = msg.SerializeToString()
        return msg_text

    def get_cntf_stream_status_msg(self):
        """
        @return:
        """
        msg = cloudgame_pb2.CGMessage()
        msg.cmdtype = cloudgame_pb2.CNtfStreamStatus
        msg.version = 6

        cntf_stream_status = cloudgame_pb2.CGCNtfStreamStatus()
        cntf_stream_status_header = cntf_stream_status.header
        cntf_stream_status_header.timestamp = int(TimeUtil.get_current_millisecond())
        cntf_stream_status_header.token = self.token
        cntf_stream_status_header.request_id = DataUtil.random_uuid4()
        cntf_stream_status.status = 1
        # AllocVPad.dpad_id = int('0')
        msg.payload = cntf_stream_status.SerializeToString()
        msg_text = msg.SerializeToString()
        allure.attach(f"flowId({self.flow_id}), uuid({self.uuid}) CNtfStreamStatus({msg_text})", name = "stream logs", attachment_type = allure.attachment_type.TEXT)
        log.info(f'flowId({self.flow_id}), uuid({self.uuid}) CNtfStreamStatus({msg_text})')

        return msg_text

    def get_cgreq_exit_game_msg(self):
        """
        @return:
        """
        msg = cloudgame_pb2.CGMessage()
        msg.cmdtype = cloudgame_pb2.ReqExitGame
        msg.version = 6

        exit_msg = cloudgame_pb2.CGReqExitGame()
        header = exit_msg.header
        header.timestamp = int(TimeUtil.get_current_millisecond())
        header.token = self.token
        header.request_id = DataUtil.random_uuid4()

        exit_msg.game_id = self.gid

        msg.payload = exit_msg.SerializeToString()
        msg_text = msg.SerializeToString()
        return msg_text

    def msg_unpack(self, recv):
        """
        消息解包
        @param recv:
        @return:
        """
        response_msg = cloudgame_pb2.CGMessage()
        response_msg.ParseFromString(recv)

        cmd_type = response_msg.cmdtype
        payload = response_msg.payload

        msg = None
        game_status = None

        if cmd_type == cloudgame_pb2.RspLaunchGame:
            event = "RspLaunchGame"
            msg = cloudgame_pb2.CGRspLaunchGame()
            msg.ParseFromString(payload)

        elif cmd_type == cloudgame_pb2.NtfGameStatus:
            """
            游戏状态通知
            游戏状态枚举(待补充)
                enum GameStatus {
                    GameStatusAllocInitial     = 0;  // 初始值 (占位)
                    GameStatusAllocSuccess     = 1;  // 游戏资源申请成功 (sdk)
                    GameStatusAllocQueuing     = 2;  // 游戏资源申请排队中 (sdk)
                    GameStatusStartSuccess     = 4;  // 游戏启动成功 (sdk)
                    GameStatusStartProcing     = 5;  // 游戏启动中 (sdk)
                    GameStatusStartFailure     = 6;  // 游戏启动失败 (sdk)
                    GameStatusCloseSuccess     = 7;  // 游戏结束 (sdk)
                    GameStatusUserUnActive     = 9;  // 用户不活跃 (sdk)
                    GameStatusAllocPrepare     = 11; // 游戏资源预分配 (sdk)
                    GameStatusIdleClosed       = 12; // 游戏闲置超时退出 (sdk)
                    GameStatusHoldClosed       = 13; // 游戏等待启动超时退出 (sdk)
                    GameStatusReadyRender      = 17; // 游戏启动成功 (sdk)
                }
            """
            # 游戏状态通知
            msg = cloudgame_pb2.CGNtfGameStatus()
            msg.ParseFromString(payload)
            event = "NtfGameStatus"
            game_status = msg.game_status

        elif cmd_type == cloudgame_pb2.RspHeartbeat:
            event = "RspHeartbeat"
            msg = cloudgame_pb2.CGRspHeartbeat()
            msg.ParseFromString(payload)

        elif cmd_type == cloudgame_pb2.NtfGameEvent:
            event = "NtfGameEvent"
            msg = cloudgame_pb2.CGNtfGameEvent()
            msg.ParseFromString(payload)

        elif cmd_type == cloudgame_pb2.RspAllocVPad:
            event = "RspAllocVPad"
            msg = cloudgame_pb2.CGRspAllocVPad()
            msg.ParseFromString(payload)
        elif cmd_type == cloudgame_pb2.CCfmStreamStatus:
            event = "CCfmStreamStatus"
            msg = cloudgame_pb2.CGCCfmStreamStatus()
            msg.ParseFromString(payload)
        elif cmd_type == cloudgame_pb2.NtfSystem:
            event = "NtfSystem"
            msg = cloudgame_pb2.CGNtfSystem()
            msg.ParseFromString(payload)
        elif cmd_type == cloudgame_pb2.CfmSystem:
            event = "CfmSystem"
            msg = cloudgame_pb2.CGCfmSystem()
            msg.ParseFromString(payload)
        elif cmd_type == cloudgame_pb2.NtfCodecEvent:
            event = "NtfCodecEvent"
            msg = cloudgame_pb2.CGNtfCodecEvent()
            msg.ParseFromString(payload)
        elif cmd_type == cloudgame_pb2.RspSetHostMethod:
            event = "RspSetHostMethod"
            msg = cloudgame_pb2.CGRspSetHostMethod()
            msg.ParseFromString(payload)
        elif cmd_type == cloudgame_pb2.NtfGameCharge:
            event = "CfmGameCharge"
            msg = cloudgame_pb2.CGNtfGameCharge()
            msg.ParseFromString(payload)
        elif cmd_type == cloudgame_pb2.GameStatusAllocQueuing:
            msg = cloudgame_pb2.QueueInfo()
            msg.ParseFromString(payload)

        elif cmd_type == cloudgame_pb2.RspExitGame:
            msg = cloudgame_pb2.CGRspExitGame()
            msg.ParseFromString(payload)

        try:
            json_obj = MessageToJson(msg)
            return cmd_type, game_status, json_obj
        except:
            return cmd_type, None, '{}'

    async def send_heartbeat(self, ws, ):
        """
        发送心跳消息
        @param ws:
        """
        if self.heartbeat_limit > 0:
            allure.attach(f"【websockets】 send heartbeat flowId({self.flow_id}), uuid({self.uuid}) heartbeat: {self.get_heart_beat_msg()}", name = "stream logs", attachment_type = allure.attachment_type.TEXT)
            log.info(f'【websockets】 send heartbeat flowId({self.flow_id}), uuid({self.uuid}) heartbeat: {self.get_heart_beat_msg()}')
            time.sleep(self.heartbeat_interval)
            await ws.send(self.get_heart_beat_msg())
            self.heartbeat_limit -= 1
        else:
            allure.attach(f"end of heartbeat flowId({self.flow_id}), uuid({self.uuid}) cardiac arrest", name = "stream logs", attachment_type = allure.attachment_type.TEXT)
            log.info(f'end of heartbeat flowId({self.flow_id}), uuid({self.uuid}) cardiac arrest')
            self.keep_alive = False

    def apply(self):
        """
        @return:
        """
        with allure.step("获取token"):
            user_token_body = post_servo_usertoken(**{"app_userid": self.app_user_id,"app_biz_type":78})
            user_token_resp = HttpClient(EnvInfo.u_servo).send_request(**user_token_body)
            self.token = DataUtil.json_extract(user_token_resp, "$.token_info.token", True)
            self.uuid = DataUtil.json_extract(user_token_resp, "$.token_info.userid", True)

        with allure.step("union apply"):
            apply_req = post_union_apply(utoken = self.token, uuid = self.uuid,**{"ugid":33945})
            apply_resp = HttpClient(EnvInfo.u_access).send_request(**apply_req)
            AssertUtils.equals_json_extract(apply_resp, "$.ret.msg", "ok", message = "用户apply资源分配失败，请检查")

        dev_type = DataUtil.json_extract(apply_req, "$.json.dev_type", True)
        protocol = "wss" if dev_type == 5 else "ws"
        access_ipv4_address = DataUtil.json_extract(apply_resp, "$.body.access_info.access_ipv4_address", True)
        access_ipv4_port = DataUtil.json_extract(apply_resp, "$.body.access_info.access_ipv4_port", True)
        apply_resp_token = DataUtil.json_extract(apply_resp, "$.body.token", True)
        dev_id = DataUtil.json_extract(apply_req, "$.json.dev_id", True)
        self.ws_path = f"{protocol}://{access_ipv4_address}:{access_ipv4_port}/v1/endpoint?token={apply_resp_token}&dev_type={dev_type}&dev_id={dev_id}"
        self.flow_id = DataUtil.json_extract(apply_resp, "$.body.flow_id", True)
        self.launch_info["apply_result"] = DataUtil.json_extract(apply_resp, "$.resp_body", True)

    def game_launch(self, path, flow_id, gid, user_id, a, v, c, ):
        """
        @param path:
        @param flow_id:
        @param gid:
        @param user_id:
        @param a:
        @param v:
        @param c:
        @return: 发起launch 游戏
        """
        launch_data = {
            "imeType": 0,
            "keyboardType": 1,
            "flowId": flow_id,
            "gId": gid,
            "startFlag": 0,
            "userId": user_id,
            "connectionType": "udpTcp",
            "forceAudioDataChannel": 0,
            "lpMultichannel": 0,
            "videoDataType": "mp4.h264",
            "supportVideoCodecProfiles": [8, 1],
            "avgBitrateratio": 100,
            "restartGame": 0,
            "deviceWidth": 0,
            "deviceHeight": 0,
            "desiredWidth": 2560,
            "desiredHeight": 1080,
            "maxFps": 60,
            "packetSize": 768,
            "clientType": "web-pc",
            "fecRepairPercent": 5,
            "clientSupportHevc": 1,
            "withHWCursor": False,
            "requestUseServerCursorData": True,
            "idrPeriod": 60,
            "videoLevel": 4,
            "vibration": True,
            "clientId": "tester",
            "forwardIp": "61.130.29.133",
            "forwardIpV6": "",
            "forwardPort": 32400,
            "forwardPortV6": 32400,
            "sdkVersion": "3.48.1",
            "joinMethod": 0,
            "targetUuid": 0,
            "aSessionId": a,
            "vSessionId": v,
            "cSessionId": c,
            "gamescreen": "landscape",
            "deviceId": "tester_xxx",
            "clientNetworkType": 40,
            "gamePlayType": 0,
            "hostMethod": 0,
            "customerType": 0,
            "supportHashCursorIco": 1
        }
        req_obj = {
            "method": "POST",
            "describe": "game launch",
            "req_body": launch_data
        }
        return HttpClient(path).send_request(**req_obj)

    def msg_handle(self, data):
        """
        消息监听,状态出来逻辑
        @param data:
        """
        cmd_type = data['cmd_type']
        game_status = data['game_status']

        if cmd_type == 400:
            # 游戏结束
            allure.attach(f"cmd_type(400), flowId({self.flow_id}), uuid({self.uuid}) payload({data['payload']})", name = "stream logs", attachment_type = allure.attachment_type.TEXT)
            log.info(f"flowId({self.flow_id}), uuid({self.uuid}) payload({data['payload']})")
            self.launch_info["exception"] = data['payload']
        else:
            payload = json.loads(data['payload'])

        if game_status == cloudgame_pb2.GameStatusAllocSuccess:
            allure.attach(f"【game state】flowId({self.flow_id}), uuid({self.uuid}), game_status(GameStatusAllocSuccess)", name = "stream logs", attachment_type = allure.attachment_type.TEXT)
            log.info(f"【game state】flowId({self.flow_id}), uuid({self.uuid}), game_status(GameStatusAllocSuccess)")
            a, v, c = None, None, None
            decode_type = None
            self.launch_info["game_resource"] = self.game_resource = payload['gameResource']

            try:
                a, v, c = self.game_resource["aSessionId"], self.game_resource["vSessionId"], self.game_resource[
                    "cSessionId"]
                decode_type = 1 if payload["decodeInfo"]["decodeType"] == "H265" else 0
            except Exception as e:
                log.error(e)
            # 游戏拉起 TODO
            # address = self.game_resource['resourceIpv4Address']
            # port = self.game_resource['resourceIpv4Portseg'] * 100 + 4
            # path = 'https://{}:{}/launch'.format(address, port, a, v, c)
            #
            # self.game_launch(path, flow_id, gid, self.user_id, a, v, c)

        elif game_status == cloudgame_pb2.GameStatusReadyRender:
            allure.attach(f"【game state】flowId({self.flow_id}), uuid({self.uuid}), game_status(GameStatusReadyRender)", name = "stream logs", attachment_type = allure.attachment_type.TEXT)
            log.info(f"【game state】flowId({self.flow_id}), uuid({self.uuid}), game_status(GameStatusReadyRender)")
            self.game_status_ready_render = True
            self.launch_success = True
            if self.mode == 1:
                # TODO 调用释放
                pass
            self.finish = 1


        elif game_status == cloudgame_pb2.GameStatusStartSuccess:
            allure.attach(f"【game state】flowId({self.flow_id}), uuid({self.uuid}), game_status(GameStatusStartSuccess)", name = "stream logs", attachment_type = allure.attachment_type.TEXT)

            log.info(f"【game state】flowId({self.flow_id}), uuid({self.uuid}), game_status(GameStatusStartSuccess)")
            self.launch_info["game_resource"] = self.game_resource = payload['gameResource']
            self.launch_success = True

        elif cmd_type == cloudgame_pb2.CCfmStreamStatus:
            self.launch_success = True

        elif cmd_type == cloudgame_pb2.NtfGameEvent:
            if payload["event"] == "GameEventJoinPvPFailure":
                self.keep_alive = False
            if payload["event"] == "GameEventJoinPvP" and self.mode == 1:  # 加入模式捕获GameEventJoinPvP视为启动成功
                self.launch_success = True
        elif cmd_type == cloudgame_pb2.CfmCodecEvent:
            pass

    async def init_connector(self, access_status = None):
        """
        @param access_status: 增加校验参数，格式[cmd_type,pay_load的keys,校验的值]，用以控制connector的启动和停止
        """
        try:
            with allure.step(f"websockets connect ws path {self.ws_path}"):
                log.info(f"websockets connect ws path {self.ws_path}")
                async with websockets.connect(self.ws_path, ssl=ssl_context) as ws:
                    launch_msg = self.get_launch_msg()
                    await ws.send(launch_msg)
                    while self.keep_alive:
                        response_msg = await ws.recv()
                        if response_msg:
                            cmd_type, game_status, json_string = self.msg_unpack(response_msg)
                            log.info(f"【websockets】  response cmd_type: {cmd_type}  game_status: {game_status} json_string: {json_string}")
                            allure.attach(f" 【websockets】  response cmd_type: {cmd_type}  game_status: {game_status} json_string: {json_string}", name = "stream logs", attachment_type = allure.attachment_type.TEXT)
                            payload = json.loads(json_string)

                            if game_status == cloudgame_pb2.GameStatusReadyRender or game_status == cloudgame_pb2.GameStatusStartSuccess:
                                self.launch_success = True

                            if game_status == cloudgame_pb2.GameStatusStartSuccess:
                                if self.mode == 1:  # 切屏要用
                                    cntf_stream_status = self.get_cntf_stream_status_msg()
                                    log.info("【确认串流】")
                                    await ws.send(cntf_stream_status)

                            if game_status == cloudgame_pb2.GameStatusAllocSuccess:
                                """
                                GameStatusAllocSuccess 同实例启动游戏状态
                                """
                                if self.mode == 1:  # 同实例要用
                                    cntf_stream_status = self.get_cntf_stream_status_msg()
                                    await ws.send(cntf_stream_status)

                            if cmd_type == cloudgame_pb2.CCfmStreamStatus:
                                pass
                                self.launch_success = True
                            # 通过事件控制连接是否断开
                            if access_status:
                                if cmd_type == access_status[0]:
                                    if payload[access_status[1]] == access_status[2]:
                                        self.keep_alive = False

                            message = {'cmd_type': cmd_type, 'game_status': game_status, 'payload': json_string}

                            self.msg_handle(message)
                            if self.launch_success:
                                await self.send_heartbeat(ws)

                            if self.finish == 1:
                                with allure.step(f"end game {self.ws_path}"):
                                    exit_msg = self.get_cgreq_exit_game_msg()
                                    await ws.send(exit_msg)
                                    allure.attach(f"【websockets】send flowId({self.flow_id}), uuid({self.uuid}), exit game msg({exit_msg})", name = "stream logs", attachment_type = allure.attachment_type.TEXT)
                                    log.info(f"【websockets】send flowId({self.flow_id}), uuid({self.uuid}), exit game msg({exit_msg})")
                                    self.finish = 0
        except Exception as e:
            """
            捕获错误模块
            """
            log.error(f"Exception err {e}")
            payload = {"code": e.code, "reason": e.reason}
            message = {'cmd_type': 400, 'game_status': None, 'payload': payload}
            self.msg_handle(message)

    def run_connector(self, access_status = None):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(self.init_connector(access_status = access_status))

    def launch(self):
        self.apply()
        self.run_connector()

    def dial_test(self):
        i = 0
        while i <= 2:
            log.debug(f"【开始下一轮测试】{i} ")
            self.apply()
            self.run_connector()
            i += 1
            self.reset()
            time.sleep(10)

    def dial_main(self):
        self.apply()
        self.run_connector()
        # self.reset()


if __name__ == '__main__':
    a = FullLinkRunner()
    # a.mode = 1
    import threading

    # a.dial_main()
    game_online = threading.Thread(target = a.dial_main)
    game_online.start()
    game_online.join()
